package se.sics.kompics.tutorial.tutorial4.PUC;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.tutorial3.RIWC.timeout.SleepTimeout;
import se.sics.kompics.tutorial.tutorial4.AC.ACImpl;
import se.sics.kompics.tutorial.tutorial4.AC.communication.ACPort;
import se.sics.kompics.tutorial.tutorial4.AC.communication.AcPropose;
import se.sics.kompics.tutorial.tutorial4.AC.communication.AcReturn;
import se.sics.kompics.tutorial.tutorial4.FELD.communication.LeaderPort;
import se.sics.kompics.tutorial.tutorial4.FELD.communication.Trust;
import se.sics.kompics.tutorial.tutorial4.PUC.event.DPostponeTimeout;
import se.sics.kompics.tutorial.tutorial4.PUC.event.Decided;
import se.sics.kompics.tutorial.tutorial4.PUC.event.UcDecide;
import se.sics.kompics.tutorial.tutorial4.PUC.event.UcPropose;

public class PaxosUniformConsensus extends ComponentDefinition {

	Negative<UniformConsensus> uc = provides(UniformConsensus.class);
	Positive<ACPort> ac = requires(ACPort.class);
	Positive<LeaderPort> eld = requires(LeaderPort.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Positive<Timer> timer = requires(Timer.class);
	
	private String[] commands;
	private int lastCommand;
	private static final Logger logger = LoggerFactory.getLogger(PaxosUniformConsensus.class);
	
	private Set<Address> neighborSet;
	private Address self;

	private Set<Integer> seenIds = new HashSet<Integer>();
	private boolean leader = false;
	private Map<Integer, Integer> proposal = new HashMap<Integer, Integer>();
	private Map<Integer, Boolean> 	proposed = new HashMap<Integer, Boolean>(), 
									decided = new HashMap<Integer, Boolean>();
	private Map<Integer, Integer> valueDecided = new HashMap<Integer, Integer>();
	
	public PaxosUniformConsensus(){
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleTrust, eld);
		subscribe(handleDecided, pp2p);
		subscribe(handleUcPropose, uc);
		subscribe(handleAcReturn, ac);
		subscribe(handleSleepTimeout, timer);
		subscribe(handlePostponeTimeout, timer);
		
	}
	
	Handler<PUC_Init> handleInit = new Handler<PUC_Init>() {

		@Override
		public void handle(PUC_Init arg0) {
			commands = arg0.getCommandScript().split(":");
			lastCommand = -1;
			
			self = arg0.getSelf();
			neighborSet = arg0.getNeighborSet();
			neighborSet.add(self);
			
		}
	};
	
	Handler<Trust> handleTrust = new Handler<Trust>() {

		@Override
		public void handle(Trust arg0) {
			if(arg0.getLeader().equals(self)){
				logger.info("I am now the leader");
				leader = true;
				for(int id : seenIds){
					tryPropose(id);
				}
			}
			else {
				logger.info("New leader trusted : process "+arg0.getLeader().getId());
				leader = false;
			}
		}
	};
	
	Handler<UcPropose> handleUcPropose = new Handler<UcPropose>() {

		@Override
		public void handle(UcPropose arg0) {
			initInstance(arg0.getId());
			proposal.put(arg0.getId(), arg0.getValue());
			tryPropose(arg0.getId());
		}
	};
	
	Handler<AcReturn> handleAcReturn = new Handler<AcReturn>() {

		@Override
		public void handle(AcReturn arg0) {
			if(arg0.getV() != ACImpl.ABORT_VALUE){
				for(Address node : neighborSet){
					trigger(new Pp2pSend(node, new Decided(self, arg0.getId(), arg0.getV())) , pp2p);
				}
				logger.info("PID"+arg0.getId()+": EP-decision: "+arg0.getV()+" -> Bcast UC-decided");
			}
			else {
				logger.info("PID"+arg0.getId()+": EP-aborted");
				proposed.put(arg0.getId(), false);
				tryPropose(arg0.getId());
			}
		}
	};
	
	Handler<Decided> handleDecided = new Handler<Decided>() {

		@Override
		public void handle(Decided arg0) {
			initInstance(arg0.getId());
			if(! decided.get(arg0.getId())){
				logger.info("PID"+arg0.getId()+": UC-decided received: "+arg0.getValue()+" -> UC-decision");
				decided.put(arg0.getId(), true);
				valueDecided.put(arg0.getId(), arg0.getValue());
				trigger(new UcDecide(arg0.getId(), arg0.getValue()), uc);
			} else {
				logger.info("PID"+arg0.getId()+": UC-decided received: "+arg0.getValue()+" -> already decided");
			}
		}
	};
	
	private void initInstance(int id){
		if(! seenIds.contains(id)){
			proposal.put(id, null);
			proposed.put(id, false);
			decided.put(id, false);
			seenIds.add(id);
		}
	}
	
	private void tryPropose(int id){
		if(		leader &&
				(proposed.get(id) == null || ! proposed.get(id)) && 
				proposal.get(id) != null){
			proposed.put(id, true);
			trigger(new AcPropose(id, proposal.get(id)), ac);
			
		}
	}
	
	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			doNextCommand();
		}
	};
	
	Handler<SleepTimeout> handleSleepTimeout = new Handler<SleepTimeout>(){

		@Override
		public void handle(SleepTimeout arg0) {
			logger.info("Waking up");
			doNextCommand();
		}
	};
	
	Handler<DPostponeTimeout> handlePostponeTimeout = new Handler<DPostponeTimeout>(){

		@Override
		public void handle(DPostponeTimeout arg0) {
			if(doCheckAllDecisionsMade()){
				doSleep(arg0.getSleepTime());
			} else {
				ScheduleTimeout requestCheck = new ScheduleTimeout(500);
				requestCheck.setTimeoutEvent(new DPostponeTimeout(requestCheck, arg0.getSleepTime()));
				trigger(requestCheck, timer);
			}
		}
	};
	
	private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {			doNextCommand();

					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}

	private void doCommand(String cmd) {
		if (cmd.startsWith("P")) {
			String[] sub = cmd.substring(1).split("-");
			logger.info("Request to UC-propose "+sub[1]+" with PID "+sub[0]);
			handleUcPropose.handle(new UcPropose(Integer.parseInt(sub[0]), Integer.parseInt(sub[1])));
			doNextCommand();
		} else if (cmd.startsWith("D")) {
			int sleepTime = Integer.parseInt(cmd.substring(1));
			if(doCheckAllDecisionsMade()){
				doSleep(sleepTime);
			} else {
				ScheduleTimeout requestCheck = new ScheduleTimeout(500);
				requestCheck.setTimeoutEvent(new DPostponeTimeout(requestCheck, sleepTime));
				trigger(requestCheck, timer);
			}
			
			
		} else if (cmd.startsWith("W")) {
			logger.info("Paxos decisions thus far :");
			for(int key : valueDecided.keySet()){
				logger.info("PID"+key+": decision done in favor of "+valueDecided.get(key));
			}
			logger.info("------------------------");
			doNextCommand();
		} else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private void doSleep(int sleepTime) {
		logger.info("Every proposal has been decided upon, sleeping for "+sleepTime+"ms");
		ScheduleTimeout requestCheck = new ScheduleTimeout(sleepTime);
		requestCheck.setTimeoutEvent(new SleepTimeout(requestCheck));
		trigger(requestCheck, timer);
	}

	private boolean doCheckAllDecisionsMade() {
		/*int nbProposals, nbDecided;
		nbProposals = proposal.size();
		nbDecided = 0;
		for (Boolean decided : this.decided.values()){
			if(decided){
				nbDecided ++;
			}
		}
		return(nbDecided == nbProposals);*/
		return(valueDecided.size() == proposal.size());
	}

	private final void doHelp() {
		logger.info("Available commands: Pi-j, Dk, W, help, X");
		logger.info("Pi-j: propose value j for Paxos instance i");
		logger.info("Dk: wait till every proposal is decided upon, then sleep for k ms");
		logger.info("W: output every decision made sorted by Paxos instance id");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
